# 此文件是一个示例的特征定义文件，用于 Feast 特征存储

from datetime import timedelta

import pandas as pd

from feast import (
    Entity,  # 定义实体类
    FeatureService,  # 定义特征服务
    FeatureView,  # 定义特征视图
    Field,  # 定义字段
    FileSource,  # 定义文件数据源
    PushSource,  # 定义推送数据源
    RequestSource,  # 定义请求数据源
)
from feast.on_demand_feature_view import on_demand_feature_view  # 导入按需特征视图装饰器
from feast.types import Float32, Float64, Int64  # 导入数据类型

# 定义一个实体，代表司机。你可以将实体视为用于获取特征的主键。
driver = Entity(name="driver", join_keys=["driver_id"])

# 从 Parquet 文件中读取数据。Parquet 格式便于本地开发模式。
# 对于生产环境，可以使用你喜欢的数据仓库，例如 BigQuery。详细信息请参阅 Feast 文档。
driver_stats_source = FileSource(
    name="driver_hourly_stats_source",
    path="data/driver_stats.parquet",  # 数据文件路径
    timestamp_field="event_timestamp",  # 时间戳字段
    created_timestamp_column="created",  # 数据创建时间戳字段
)

# 我们的 Parquet 文件包含示例数据，包括 driver_id 列、时间戳和三个特征列。
# 在这里，我们定义一个特征视图（Feature View），它将允许我们在线提供这些数据给模型。
driver_stats_fv = FeatureView(
    # 此特征视图的唯一名称。在同一个项目中，两个特征视图不能有相同的名称。
    name="driver_hourly_stats",
    entities=[driver],  # 关联的实体
    ttl=timedelta(days=1),  # 特征数据的过期时间
    # 下面定义的特征列表既充当特征的模式，用于将特征物化到存储中，
    # 也用于在构建训练数据集或提供特征时作为引用。
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64, description="Average daily trips"),
    ],
    online=True,  # 是否用于在线服务
    source=driver_stats_source,  # 数据源
    # 标签是用户定义的键值对，附加到每个特征视图上
    tags={"team": "driver_performance"},
)

# 定义一个请求数据源，它编码了仅在请求时可用的特征/信息（例如，用户发起的 HTTP 请求的一部分）
input_request = RequestSource(
    name="vals_to_add",
    schema=[
        Field(name="val_to_add", dtype=Int64),
        Field(name="val_to_add_2", dtype=Int64),
    ],
)

# 作用：使用 @on_demand_feature_view 装饰器定义了一个按需特征视图 transformed_conv_rate。
# 该特征视图以 driver_stats_fv 和 input_request 作为数据来源， 生成新的特征 conv_rate_plus_val1 和 conv_rate_plus_val2。
@on_demand_feature_view(
    sources=[driver_stats_fv, input_request],  # 数据来源，包括特征视图和请求源
    schema=[
        Field(name="conv_rate_plus_val1", dtype=Float64),
        Field(name="conv_rate_plus_val2", dtype=Float64),
    ],
)
def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:
    # 定义特征计算逻辑
    df = pd.DataFrame()
    df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
    df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"]
    return df

# 将特征分组到一个模型版本中
driver_activity_v1 = FeatureService(
    name="driver_activity_v1",
    features=[
        driver_stats_fv[["conv_rate"]],  # 从特征视图中选择特定的特征
        transformed_conv_rate,  # 选择特征视图中的所有特征
    ],
)

driver_activity_v2 = FeatureService(
    name="driver_activity_v2",
    features=[driver_stats_fv, transformed_conv_rate],
)

# 定义一种将数据推送（可用于离线、在线或两者）到 Feast 的方式。
driver_stats_push_source = PushSource(
    name="driver_stats_push_source",
    batch_source=driver_stats_source,  # 批量数据源
)

# 定义一个与上面稍有修改的特征视图，其中数据源已更改为推送源。
# 这允许新鲜的特征直接被推送到此特征视图的在线存储中。
driver_stats_fresh_fv = FeatureView(
    name="driver_hourly_stats_fresh",
    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
    ],
    online=True,
    source=driver_stats_push_source,  # 数据源更改为推送源
    tags={"team": "driver_performance"},
)

# 定义一个按需特征视图，可以基于现有的特征视图和请求源特征生成新特征
@on_demand_feature_view(
    sources=[driver_stats_fresh_fv, input_request],  # 依赖于新鲜版本的特征视图
    schema=[
        Field(name="conv_rate_plus_val1", dtype=Float64),
        Field(name="conv_rate_plus_val2", dtype=Float64),
    ],
)
def transformed_conv_rate_fresh(inputs: pd.DataFrame) -> pd.DataFrame:
    # 定义特征计算逻辑
    df = pd.DataFrame()
    df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
    df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"]
    return df

driver_activity_v3 = FeatureService(
    name="driver_activity_v3",
    features=[driver_stats_fresh_fv, transformed_conv_rate_fresh],
)